package com.ds.infrastructure.hyperspace.container.endpoint.destination.resolver;

import com.google.common.collect.Sets;
import com.ds.infrastructure.hyperspace.common.config.TargetNodeConfig;
import com.ds.infrastructure.hyperspace.container.endpoint.connector.HyperspaceConnector;
import com.ds.infrastructure.hyperspace.container.internal.NettyContainer;
import com.ds.infrastructure.hyperspace.container.internal.util.AddressesUtil;
import io.netty.channel.EventLoop;
import lombok.Getter;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.Collections;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author: yaozh
 * @Description:
 */
public abstract class BaseResolverProvider implements ResolverProvider {

    private static final int MAX_REWIND_SIZE = 1000000;
    private AtomicInteger currentCount = new AtomicInteger(0);
    @Getter
    protected TargetNodeConfig targetNodeConfig;
    private volatile Set<InetSocketAddress> nodeSet = new HashSet<>(10);
    private volatile InetSocketAddress[] nodeArray = {};
    private NettyContainer nettyContainer;
    @Getter
    private String targetEndpointNamespace;
    @Getter
    private volatile boolean isClosed;

    protected BaseResolverProvider(NettyContainer nettyContainer, TargetNodeConfig targetNodeConfig, String targetEndpointNamespace) {
        this.targetNodeConfig = targetNodeConfig;
        this.nettyContainer = nettyContainer;
        this.targetEndpointNamespace = targetEndpointNamespace;
    }

    @Override
    public InetSocketAddress doSelect() {
        try {
            return getNode();
        } catch (IndexOutOfBoundsException ex) {
            return null;
        }
    }

    private InetSocketAddress getNode() {
        int nodeSize = nodeArray.length;
        if (nodeSize == 0) {
            return null;
        }
        if (nodeSize == 1) {
            return nodeArray[0];
        }
        int index = currentCount.incrementAndGet();
        if (index > MAX_REWIND_SIZE) {
            index = 0;
            currentCount.set(index);
        }
        return nodeArray[index % nodeSize];
    }

    @Override
    public boolean isAvailable(SocketAddress address) {
        return nodeSet.contains(address);
    }

    @Override
    public void reloadNode(Set<InetSocketAddress> newNodeSet) {
        Sets.SetView<InetSocketAddress> oldDifference = Sets.difference(nodeSet, newNodeSet);
        Sets.SetView<InetSocketAddress> newDifference = Sets.difference(newNodeSet, nodeSet);
        if (oldDifference.size() > 0 || newDifference.size() > 0) {
            updateNodeSet(newNodeSet);
        }
    }

    private void updateNodeSet(Set<InetSocketAddress> newNodeSet) {
        Map<EventLoop, HyperspaceConnector> hyperspaceConnectorMap =
                nettyContainer.getHyperspaceConnectorMap();
        for (HyperspaceConnector hyperspaceConnector : hyperspaceConnectorMap.values()) {
            hyperspaceConnector.eventLoop().submit(() -> {
                if (isClosed) {
                    return;
                }
                nodeArray = newNodeSet.toArray(new InetSocketAddress[newNodeSet.size()]);
                nodeSet = newNodeSet;
                removeTargetNodeStatusMetrics(newNodeSet);
            });
        }
    }

    private void removeTargetNodeStatusMetrics(Set<InetSocketAddress> newNodeSet) {
        Set<String> targetNodeSet = new HashSet<>(newNodeSet.size());
        for (InetSocketAddress inetSocketAddress : newNodeSet) {
            targetNodeSet.add(AddressesUtil.getIpAndPort(inetSocketAddress));
        }
        nettyContainer.getMetricsManager().removeTargetNodeStatusGauge(targetEndpointNamespace, targetNodeSet);
    }

    @Override
    public Set<InetSocketAddress> getAllNode() {
        return nodeSet;
    }

    @Override
    public void shutdown() {
        cancel();
        clearData();
    }

    private void clearData() {
        Map<EventLoop, HyperspaceConnector> hyperspaceConnectorMap =
                nettyContainer.getHyperspaceConnectorMap();
        for (HyperspaceConnector hyperspaceConnector : hyperspaceConnectorMap.values()) {
            hyperspaceConnector.eventLoop().submit(() -> {
                isClosed = true;
                nodeArray = new InetSocketAddress[]{};
                nodeSet = Collections.emptySet();
                removeTargetNodeStatusMetrics(Collections.emptySet());
            });
        }
    }
}
